package io.kiki.sba.registry.server.clientInterface.metadata;

import com.google.common.util.concurrent.Futures;
import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.appmeta.InterfaceMapping;
import io.kiki.sba.registry.common.model.metaserver.MultiClusterSyncInfo;
import io.kiki.sba.registry.common.model.store.AppRevision;
import io.kiki.sba.registry.server.clientInterface.bootstrap.ExecutorManager;
import io.kiki.sba.registry.server.clientInterface.bootstrap.MultiClusterSessionServerConfig;
import io.kiki.sba.registry.store.api.config.DefaultCommonConfig;
import io.kiki.sba.registry.store.api.meta.MultiClusterSyncRepository;
import io.kiki.sba.registry.store.api.repository.AppRevisionRepository;
import io.kiki.sba.registry.store.api.repository.InterfaceAppsRepository;
import io.kiki.sba.registry.util.AtomicMap;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.LoopExecuteTask;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

public class MetadataCacheRegistry {

    private static final Logger logger = LoggerFactory.getLogger(MetadataCacheRegistry.class);
    private final MetadataRegisterWorker metadataRegisterWorker = new MetadataRegisterWorker();
    private final MultiSyncInfoReloadWorker multiSyncInfoReloadWorker = new MultiSyncInfoReloadWorker();
    @Autowired
    private AppRevisionRepository appRevisionRepository;
    @Autowired
    private InterfaceAppsRepository interfaceAppsRepository;
    @Autowired
    private MultiClusterSyncRepository multiClusterSyncRepository;
    @Autowired
    private DefaultCommonConfig defaultCommonConfig;
    @Autowired
    private MultiClusterSessionServerConfig multiClusterSessionServerConfig;
    @Autowired
    private ExecutorManager executorManager;
    private AtomicMap<String, AppRevision> registerTask = new AtomicMap<>();
    private AtomicReference<Set<String>> pushEnableSet = new AtomicReference<>();

    @PostConstruct
    public void init() {
        ConcurrentUtils.createDaemonThread("MetadataRegisterWorker", metadataRegisterWorker).start();
        ConcurrentUtils.createDaemonThread("MultiSyncInfoReloadWorker", multiSyncInfoReloadWorker).start();
    }

    public Set<String> getPushEnableDataCenters() {
        Set<String> set = pushEnableSet.get();
        return set == null ? Collections.EMPTY_SET : set;
    }

    public void register(AppRevision appRevision) {
        registerTask.put(appRevision.getRevision(), appRevision);
        metadataRegisterWorker.wakeup();
    }

    // retry not wakeup
    public void registerNotWake(AppRevision appRevision) {
        registerTask.put(appRevision.getRevision(), appRevision);
    }

    private Tuple<Set<String>, Set<String>> getMetadataDataCenters() {
        Set<MultiClusterSyncInfo> set = getSyncDataCenters();

        Set<String> syncDataCenters = new HashSet<>();
        Set<String> syncEnableDataCenters = new HashSet<>();
        syncDataCenters.add(defaultCommonConfig.getDefaultClusterId());

        for (MultiClusterSyncInfo multiClusterSyncInfo : set) {
            syncDataCenters.add(multiClusterSyncInfo.getRemoteDataCenter());
            if (multiClusterSyncInfo.isEnablePush()) {
                syncEnableDataCenters.add(multiClusterSyncInfo.getRemoteDataCenter());
            }
        }
        return new Tuple<>(syncDataCenters, syncEnableDataCenters);
    }

    private Set<MultiClusterSyncInfo> getSyncDataCenters() {

        return multiClusterSyncRepository.queryLocalMultiClusterSyncInfoSet();
    }

    public void startSynced() {
        logger.info("metadata cache enter startSynced.");
        Tuple<Set<String>, Set<String>> tuple = getMetadataDataCenters();

        appRevisionRepository.setDataCenters(tuple.o1);
        appRevisionRepository.startSynced();

        interfaceAppsRepository.setDataCenters(tuple.o1);
        interfaceAppsRepository.startSynced();

        pushEnableSet.set(tuple.o2);
        logger.info("metadata cache finish startSynced.");
    }

    public void waitSynced() {
        appRevisionRepository.waitSynced();
        interfaceAppsRepository.waitSynced();
    }

    public boolean heartbeat(String revision) {
        return appRevisionRepository.heartbeat(revision);
    }

    public InterfaceMapping getAppNames(String dataInfoId) {
        return interfaceAppsRepository.getAppNames(dataInfoId);
    }

    public AppRevision getRevision(String revision) {
        return appRevisionRepository.queryRevision(revision);
    }

    private Map<String, Future<Boolean>> doRegister(Map<String, AppRevision> registers) {
        Map<String, Future<Boolean>> map = new HashMap<>(registers.size());
        for (AppRevision appRevision : registers.values()) {

            try {
                Future<Boolean> future = executorManager.getAppRevisionRegisterExecutor().submit(() -> {
                    try {
                        appRevisionRepository.register(appRevision);
                        return true;
                    } catch (Throwable t) {
                        logger.error("[AppRevision]register error, AppRevision:{}", appRevision, t);
                        return false;
                    }
                });
                map.put(appRevision.getRevision(), future);
            } catch (Throwable t) {
                logger.error("[AppRevision]submit register error, AppRevision:{}", appRevision, t);
                map.put(appRevision.getRevision(), Futures.immediateFuture(false));
            }
        }
        return map;
    }


    private class MetadataRegisterWorker extends WakeUpLoopExecuteTask {

        @Override
        public void _execute_() {
            Map<String, AppRevision> map = registerTask.getAndReset();

            if (CollectionUtils.isEmpty(map)) {
                return;
            }

            Map<String, Future<Boolean>> futures = doRegister(map);

            for (Entry<String, Future<Boolean>> entry : futures.entrySet()) {
                try {
                    Boolean success = entry.getValue().get(3000, TimeUnit.MILLISECONDS);

                    if (success == null || !success) {
                        String revision = entry.getKey();
                        logger.info("register fail, retry ro register revision:{}", revision);
                        registerNotWake(map.get(revision));
                    }
                } catch (Throwable throwable) {
                    String revision = entry.getKey();
                    registerNotWake(map.get(revision));
                    logger.error("[AppRevision]do register error, retry next time. revision:{}", revision, throwable);
                }
            }
        }

        @Override
        public int getWaitingMillis() {
            return 100;
        }
    }

    private class MultiSyncInfoReloadWorker extends LoopExecuteTask {

        @Override
        public void _execute_() {
            Tuple<Set<String>, Set<String>> tuple = getMetadataDataCenters();
            appRevisionRepository.setDataCenters(tuple.o1);
            interfaceAppsRepository.setDataCenters(tuple.o1);

            pushEnableSet.set(tuple.o2);
        }

        @Override
        public void _wait_() {
            ConcurrentUtils.sleepUninterruptibly(multiClusterSessionServerConfig.getMultiClusterConfigReloadSecs(), TimeUnit.SECONDS);
        }
    }
}
